/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.kafka.image;

import org.apache.kafka.image.node.MetadataImageNode;
import org.apache.kafka.image.writer.ImageWriter;
import org.apache.kafka.image.writer.ImageWriterOptions;
import org.apache.kafka.raft.OffsetAndEpoch;

import java.util.Objects;


/**
 * The broker metadata image.
 *
 * This class is thread-safe.
 */
public final class MetadataImage {
    public static final MetadataImage EMPTY = new MetadataImage(
        MetadataProvenance.EMPTY,
        FeaturesImage.EMPTY,
        ClusterImage.EMPTY,
        TopicsImage.EMPTY,
        ConfigurationsImage.EMPTY,
        ClientQuotasImage.EMPTY,
        ProducerIdsImage.EMPTY,
        AclsImage.EMPTY,
        ScramImage.EMPTY,
        DelegationTokenImage.EMPTY,
        S3StreamsMetadataImage.EMPTY,
        S3ObjectsImage.EMPTY,
        KVImage.EMPTY);

    private final MetadataProvenance provenance;

    private final FeaturesImage features;

    private final ClusterImage cluster;

    private final TopicsImage topics;

    private final ConfigurationsImage configs;

    private final ClientQuotasImage clientQuotas;

    private final ProducerIdsImage producerIds;

    private final AclsImage acls;

    private final ScramImage scram;

    private final DelegationTokenImage delegationTokens;

    // AutoMQ for Kafka inject start

    private final S3StreamsMetadataImage streamMetadata;

    private final S3ObjectsImage objectsMetadata;

    private final KVImage kv;
    // AutoMQ for Kafka inject end

    public MetadataImage(
        MetadataProvenance provenance,
        FeaturesImage features,
        ClusterImage cluster,
        TopicsImage topics,
        ConfigurationsImage configs,
        ClientQuotasImage clientQuotas,
        ProducerIdsImage producerIds,
        AclsImage acls,
        ScramImage scram,
        DelegationTokenImage delegationTokens,
        S3StreamsMetadataImage streamMetadata,
        S3ObjectsImage s3ObjectsImage,
        KVImage kvImage
    ) {
        this.provenance = provenance;
        this.features = features;
        this.cluster = cluster;
        this.topics = topics;
        this.configs = configs;
        this.clientQuotas = clientQuotas;
        this.producerIds = producerIds;
        this.acls = acls;
        this.scram = scram;
        this.delegationTokens = delegationTokens;
        this.streamMetadata = streamMetadata;
        this.objectsMetadata = s3ObjectsImage;
        this.kv = kvImage;
    }

    public boolean isEmpty() {
        return features.isEmpty() &&
            cluster.isEmpty() &&
            topics.isEmpty() &&
            configs.isEmpty() &&
            clientQuotas.isEmpty() &&
            producerIds.isEmpty() &&
            acls.isEmpty() &&
            scram.isEmpty() &&
            delegationTokens.isEmpty() &&
            streamMetadata.isEmpty() &&
            objectsMetadata.isEmpty() &&
            kv.isEmpty();
    }

    public MetadataProvenance provenance() {
        return provenance;
    }

    public OffsetAndEpoch highestOffsetAndEpoch() {
        return new OffsetAndEpoch(provenance.lastContainedOffset(), provenance.lastContainedEpoch());
    }

    public long offset() {
        return provenance.lastContainedOffset();
    }

    public FeaturesImage features() {
        return features;
    }

    public ClusterImage cluster() {
        return cluster;
    }

    public TopicsImage topics() {
        return topics;
    }

    public ConfigurationsImage configs() {
        return configs;
    }

    public ClientQuotasImage clientQuotas() {
        return clientQuotas;
    }

    public ProducerIdsImage producerIds() {
        return producerIds;
    }

    public AclsImage acls() {
        return acls;
    }

    public ScramImage scram() {
        return scram;
    }

    public DelegationTokenImage delegationTokens() {
        return delegationTokens;
    }

    // AutoMQ for Kafka inject start
    public S3StreamsMetadataImage streamsMetadata() {
        return streamMetadata;
    }

    public S3ObjectsImage objectsMetadata() {
        return objectsMetadata;
    }

    public KVImage kv() {
        return kv;
    }
    // AutoMQ for Kafka inject end

    public void write(ImageWriter writer, ImageWriterOptions options) {
        // Features should be written out first so we can include the metadata.version at the beginning of the
        // snapshot
        features.write(writer, options);
        cluster.write(writer, options);
        topics.write(writer, options);
        configs.write(writer, options);
        clientQuotas.write(writer, options);
        producerIds.write(writer, options);
        acls.write(writer, options);
        scram.write(writer, options);
        delegationTokens.write(writer, options);
        // AutoMQ for Kafka inject start
        streamMetadata.write(writer, options);
        objectsMetadata.write(writer, options);
        kv.write(writer, options);
        // AutoMQ for Kafka inject end
        writer.close(true);
    }

    @Override
    public boolean equals(Object o) {
        if (o == null || !o.getClass().equals(this.getClass())) return false;
        MetadataImage other = (MetadataImage) o;
        return provenance.equals(other.provenance) &&
            features.equals(other.features) &&
            cluster.equals(other.cluster) &&
            topics.equals(other.topics) &&
            configs.equals(other.configs) &&
            clientQuotas.equals(other.clientQuotas) &&
            producerIds.equals(other.producerIds) &&
            acls.equals(other.acls) &&
            scram.equals(other.scram) &&
            delegationTokens.equals(other.delegationTokens) &&
            streamMetadata.equals(other.streamMetadata) &&
            objectsMetadata.equals(other.objectsMetadata) &&
            kv.equals(other.kv);
    }

    @Override
    public int hashCode() {
        return Objects.hash(
            provenance,
            features,
            cluster,
            topics,
            configs,
            clientQuotas,
            producerIds,
            acls,
            scram,
            delegationTokens,
            streamMetadata,
            objectsMetadata,
            kv);
    }

    @Override
    public String toString() {
        return new MetadataImageNode(this).stringify();
    }

    // AutoMQ inject start
    public void retain() {
        streamsMetadata().retain();
        objectsMetadata().retain();
        kv().retain();
    }

    public void release() {
        streamsMetadata().release();
        objectsMetadata().release();
        kv().release();
    }
    // AutoMQ inject end
}
